package com.fm.mqtt;

import com.fm.entity.Equipment;
import com.fm.entity.Product;
import com.fm.entity.TopicLog;
import com.fm.service.EquipmentService;
import com.fm.service.ProductService;
import com.fm.service.TopicLogService;
import com.fm.utils.JSONUtils;
import com.fm.utils.SpringUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.context.ApplicationContext;
import org.springframework.messaging.simp.SimpMessagingTemplate;

import java.util.Date;

/**
 * Created by Administrator on 2018/8/7 0007.
 */
public class MqttPushClient {
//
    private static final String HOST = "tcp://192.168.0.66:61613";
    private static final String CLIENT_ID = "mqttjs_e8022a4d0b";
    private static final String USER_NAME = "admin";
    private static final String PASSWORD = "password";
//    private static final String HOST = "tcp://192.168.0.57:1883";
//    private static final String CLIENT_ID = "mqttjs_e8022a4d0b";
//    private static final String USER_NAME = "hry";
//    private static final String PASSWORD = "hry";
    private static final Integer TIMEOUT = 10;
    private static final Integer KEEPALIVE = 20;

    private ApplicationContext applicationContext = SpringUtils.getApplicationContext();
    private EquipmentService equipmentService = applicationContext.getBean(EquipmentService.class);
    private TopicLogService topicLogService = applicationContext.getBean(TopicLogService.class);

    private MqttClient client;

    private static volatile MqttPushClient mqttPushClient = null;

    public static MqttPushClient getInstance(){

        if(null == mqttPushClient){
            synchronized (MqttPushClient.class){
                if(null == mqttPushClient){
                    mqttPushClient = new MqttPushClient();
                }
            }

        }
        return mqttPushClient;

    }

    private MqttPushClient() {
        connect();
    }

    private void connect(){
        try {
            client = new MqttClient(HOST, CLIENT_ID, new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(false);
            options.setUserName(USER_NAME);
            options.setPassword(PASSWORD.toCharArray());
            options.setConnectionTimeout(TIMEOUT);
            options.setKeepAliveInterval(KEEPALIVE);
            try {
                client.setCallback(new PushCallback());
                client.connect(options);
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 发布，默认qos为0，非持久化
     * @param topic
     * @param pushMessage
     */
    public void publish(String topic,PushPayload pushMessage){
        publish(0, false, topic, pushMessage);
    }

    /**
     * 调试单个topic
     * @param topic
     * @param data
     * @param qos
     */
    public void publish(String topic, String data, Integer qos) {
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setPayload(data.getBytes());
        message.setRetained(false);
        MqttTopic mTopic = client.getTopic(topic);
        MqttDeliveryToken token;
        try {
            token = mTopic.publish(message);
            token.waitForCompletion();
        } catch (MqttPersistenceException e) {
            e.printStackTrace();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 通用的发送topic
     * @param eqId
     * @param topic
     * @param link
     * @return
     */
    public TopicLog publish(Integer eqId, String topic, String link) {
        MqttMessage message = new MqttMessage();
        message.setQos(1);
        message.setRetained(false);
//        String json = JSONUtils.toJSONString(link);
        message.setPayload(link.getBytes());
        MqttTopic mTopic = client.getTopic(topic);
        Equipment equipment = equipmentService.selectOneByExample(eqId).get();
        if(null == mTopic){
            System.out.println("topic not exist");
        }
        MqttDeliveryToken token;
        try {
            token = mTopic.publish(message);
            token.waitForCompletion();
            return writeLog(equipment.getId(),equipment.getProductId(),3,1, link);
        } catch (MqttPersistenceException e) {
            e.printStackTrace();
            return writeLog(equipment.getId(),equipment.getProductId(),3,0, e.getMessage());
        } catch (MqttException e) {
            e.printStackTrace();
            return writeLog(equipment.getId(),equipment.getProductId(),3,0, e.getMessage());
        }
    }
    /**
     * 发布
     * @param qos
     * @param retained
     * @param topic
     * @param pushMessage
     */
    public void publish(int qos,boolean retained,String topic,PushPayload pushMessage){
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage.toString().getBytes());
        MqttTopic mTopic = client.getTopic(topic);
        if(null == mTopic){
            System.out.println("topic not exist");
        }
        MqttDeliveryToken token;
        try {
            token = mTopic.publish(message);
            token.waitForCompletion();
        } catch (MqttPersistenceException e) {
            e.printStackTrace();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public void unsubscribe(String topics []) {
        try {
            client.unsubscribe(topics);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public void subscribeSome(String topics []) {
        try {
            client.subscribe(topics);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 订阅某个主题，qos默认为0
     * @param topic
     */
    public void subscribe(String topic){
        subscribe(topic,0);
    }

    /**
     * 订阅某个主题
     * @param topic
     * @param qos
     */
    public void subscribe(String topic,int qos){
        try {
            client.subscribe(topic, qos);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 写日志
     * @param equipmentId
     * @param productId
     * @param type
     * @param status
     * @param content
     */
    public TopicLog writeLog(Integer equipmentId, Integer productId, Integer type, Integer status, String content) {
        TopicLog log = new TopicLog();
        log.setEquipmentId(equipmentId);
        log.setProductId(productId);
        log.setType(type);
        log.setStatus(status);
        log.setContent(content);
        log.setCreateTime(new Date());
        topicLogService.insert(log);
        return log;
    }

    public static void main(String[] args) throws Exception {
        String kdTopic = "/sys/aK1Uk456Jmb/hand/thing/service/property/set";
        String data = "{ \"PowerSwitch\": 1 }";
        MqttPushClient.getInstance().publish(kdTopic, data, 0);
    }
}
